package cn.jly.flink.quickstart;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;

import java.util.ArrayList;
import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lanyangji
 * @create 2020-08-31 15:04
 */
public class SqlWordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tabEnv = BatchTableEnvironment.create(env);

        String words = "hello world flink flink spark spark spark";
        String[] split = words.split("\\W+");
        ArrayList<WordWithCount> wordWithCounts = new ArrayList<>();
        Arrays.stream(split).forEach(s -> wordWithCounts.add(new WordWithCount(s, 1L)));

        DataSource<WordWithCount> input = env.fromCollection(wordWithCounts);
        // dataSet转table，并指定列名
        Table table = tabEnv.fromDataSet(input, $("word").as("word"), $("count").as("frequency"));
        table.printSchema();
        // 注册一个表
        tabEnv.createTemporaryView("wordcount", table);
        // 执行查询
        Table table2 = tabEnv.sqlQuery("select word, sum(frequency) as `count` from wordcount group by word");
        // table转为dataSet
        DataSet<WordWithCount> wordWithCountDataSet = tabEnv.toDataSet(table2, WordWithCount.class);
        wordWithCountDataSet.printToErr();
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordWithCount {
        private String word;
        private Long count;

        @Override
        public String toString() {
            return "(" + word + ", " + count + ")";
        }
    }
}
